package com.j.lemon.learn.pulsar;

import org.apache.pulsar.client.api.*;

public class DeadLetterConsumer {
    public static void main(String[] args) throws PulsarClientException {
        PulsarClient pulsarClient = PulsarUtils.getClient();
        ReaderBuilder<byte[]> readerBuilder = pulsarClient.newReader();

        pulsarClient.newConsumer()
                .subscriptionType(SubscriptionType.Shared)
                .subscriptionName("deadLetterTopic")
                .topic("persistent://4a/audit/deadLetter")
                .consumerName("dead")
                .messageListener(new MessageListener<byte[]>() {
                    @Override
                    public void received(Consumer<byte[]> consumer, Message<byte[]> msg) {
                        System.out.println(consumer.getConsumerName()+"接收到一条死信消息："+new String(msg.getValue()));
                        try {
                            consumer.acknowledge(msg);
                        } catch (PulsarClientException e) {
                            e.printStackTrace();
                        }
                    }
                })
                .subscribe();
    }
}
